package com.syp.test.clustercacheflush.util.zkflushcache.handler;

import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.ZkConnection;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSON;
import com.syp.test.clustercacheflush.util.Constants;
import com.syp.test.clustercacheflush.util.HexStringUtil;
import com.syp.test.clustercacheflush.util.zkflushcache.ZkCacheProperties;
import com.syp.test.clustercacheflush.util.zkflushcache.ZkCacheSerializer;
import com.syp.test.clustercacheflush.util.zkflushcache.ZkCacheSyncNotify;

import lombok.extern.slf4j.Slf4j;

@Component
@Slf4j
public class CacheSyncNotifyMater implements InitializingBean, DisposableBean {

	@Autowired
	private ZkCacheProperties properties;
	@Autowired
	private ZkCacheSerializer zkCacheSerializer;

	private ZkClient zk;

	public void writeCacheSyncNotify(ZkCacheSyncNotify zkNotify) {
		log.info("notify:{}", zkNotify);
		
		String jsonString = JSON.toJSONString(zkNotify);
		String chldName = HexStringUtil.convert2HexStr(jsonString);

		StringBuilder sb = new StringBuilder();
		sb.append(Constants.CLUSTER_CACHE_FLUSH_ZK_PATH + "/");
		sb.append(chldName);
		
		zk.delete(sb.toString());
		zk.createEphemeral(sb.toString());
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		ZkConnection zkconn = null;
		if (properties.getSessionTimeout() > 0) {
			zkconn = new ZkConnection(properties.getZkServers(), properties.getSessionTimeout());
		} else {
			zkconn = new ZkConnection(properties.getZkServers());
		}

		if (properties.getConnectionTimeout() > 0 && properties.getOperationRetryTimeout() > 0) {
			zk = new ZkClient(zkconn, properties.getConnectionTimeout(), zkCacheSerializer, properties.getOperationRetryTimeout());
		} else if (properties.getConnectionTimeout() > 0) {
			zk = new ZkClient(zkconn, properties.getConnectionTimeout());
		} else {
			zk = new ZkClient(zkconn);
		}

		boolean blnExists = zk.exists(Constants.APP_ZK_ROOT_PATH);
		if (!blnExists) {
			zk.createPersistent(Constants.APP_ZK_ROOT_PATH);
		}
		boolean blnCacheNodeExists = zk.exists(Constants.CLUSTER_CACHE_FLUSH_ZK_PATH);
		if (!blnCacheNodeExists) {
			zk.createPersistent(Constants.CLUSTER_CACHE_FLUSH_ZK_PATH);
		}
	}

	@Override
	public void destroy() throws Exception {
		zk.close();
	}

}